package com.alili_tv.system.common;

import com.alili_tv.system.entity.Video;
import com.alili_tv.system.entity.VideoRank;
import com.alili_tv.system.exception.BusinessException;
import com.alili_tv.system.service.VideoRankService;
import com.alili_tv.system.service.VideoService;
import com.alili_tv.system.utils.CommonUtils;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Stream;

/**
 * 基于redis的排行榜
 * 设计的排行榜分为三种, 每日排行榜, 每周排行榜, 每月排行榜
 * 同时排行榜的热度不能持续不断增加, 否则新发布的视频很难超越旧视频
 * --- 基于队列的思想, 将视频的热度比作一条队列, 以每日排行榜举例, 其队列长度为24,
 * --- 每个单元就为每小时的热度增量, 当队列占满时, 从队尾剔除一小时增量, 具体实现如下
 * <p>
 * 每日排行榜:
 * 使用redis的sort set类型, value为视频的id, score为视频的热度
 * 使用 springboot 的定时器, 每隔一个小时记录一下在榜视频的热度增量, 将热度增量存入数据库中
 * 同时, 从数据库查出24小时前的一小时热度增量, 在redis中减去该热度增量
 * <p>
 * 每周排行榜的降热, 则是根据每日排行榜的数据库实现, 直接从数据库中统计7天前的热度增量再从redis中减去即可, 定时器间隔设置为1天
 * 每月排行榜同理
 * <p>
 * 相关操作: 播放+5, 评论+10, 弹幕+10, 点赞+10, 收藏+10, *视频时长/10
 * <p>
 * 以上只能实现总榜的功能, 我们还需实现按分类查询排行榜的功能
 * 保存的value格式: {分区id}(视频id), 这样就可以通过redis的模糊匹配 来查询到 指定的分区下的排行榜
 * *：通配任意多个字符
 * ?：通配单个字符
 * []：通配括号内的某一个字符
 * <p>
 * eg: 查询总榜
 * -> range = ops.reverseRangeWithScores(key, start, end);
 * <p>
 * eg: 查询分类为3的视频热榜
 * -> scanOptions = ScanOptions.scanOptions().match("*{3}*").build();
 * -> result = ops.scan(key, scanOptions);
 * <p>
 * eg: 查询视频id为5的视频
 * -> scanOptions = ScanOptions.scanOptions().match("*(5)*").build();
 * -> result = ops.scan(key, scanOptions);
 */
@Component
public class RedisRank {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private VideoRankService videoRankService;

    @Resource
    private VideoService videoService;

    // 每日排行榜的key名
    private final String DAILY_RANKING = "daily_ranking";

    // 每周排行榜
    private final String WEEK_RANKING = "week_ranking";

    // 每月排行榜
    private final String MONTH_RANKING = "month_ranking";

    public RedisRank() {
        // 清空
    }

    /**
     * 定时器, 每小时执行一次
     */
    @Scheduled(cron = "0 0 * * * ?")
    private void hourTimer() {

        System.out.println(LocalDateTime.now() + ", 执行了每小时定时器");

        // 获取当前排行榜的名单
        List<VideoRank> rankDaily = getRankList(DAILY_RANKING, 100);

        // 获取24小时前的热度值增量
        LocalDateTime now = LocalDateTime.now();
        List<VideoRank> incr = videoRankService.getIncrYesterday(now);

        // 去除24小时前的热度, 同时收集当前需要更新的热度视频
        List<VideoRank> saveList = new ArrayList<>();
        if (incr != null) {
            for (VideoRank rank : incr) {
                // 去除24小时前的热度增量, 返回去除后的热度值
                Double scoreEnd = incrScoreDaily(rank.getVideoId(), 0 - rank.getIncr());
                // 若去除后热度值小于等于0, 就不用再向数据库更新
                if (scoreEnd <= 0) {
                    continue;
                }
                Double scoreFront = scoreEnd + rank.getIncr();

                rank.setScoreEnd(scoreEnd);
                rank.setScoreFront(scoreFront);
                saveList.add(rank);
            }
        }

        // 将数据库已存在的数据和redis中的数据进行合并 去重
        // redis中可能存在新的数据, 需要将其添加至数据库中
        // 重复数据应该优先保存数据库中筛选的数据, 因为需要保存更新后的数据(上段代码已实现)
        saveList = Stream.concat(saveList.stream(), rankDaily.stream())
                .distinct()
                .filter(CommonUtils.distinctByKey(VideoRank::getVideoId))
                .peek(item -> item.setCreateTime(now))
                .toList();

        // 保存
        videoRankService.saveBatch(saveList);
    }

    /**
     * 定时器, 每天执行一次, 更新周榜和月榜
     */
    @Scheduled(cron = "0 0 0 * * ?")
    private void dailyTimer() {
        System.out.println(LocalDateTime.now() + ", 执行了每日定时器");

        LocalDateTime now = LocalDateTime.now();
        // 获取7天前的热度增量
        List<VideoRank> incrWeek = videoRankService.getIncrAtOneDay(now.plusWeeks(-1));
        if (incrWeek != null) {
            incrWeek.forEach(item -> incrScoreWeek(item.getVideoId(), 0 - item.getIncr()));
        }

        // 获取三十天前的热度增量
        List<VideoRank> incrMonth = videoRankService.getIncrAtOneDay(now.plusDays(-30));
        if (incrMonth != null) {
            incrMonth.forEach(item -> incrScoreMonth(item.getVideoId(), 0 - item.getIncr()));
        }
    }

    /**
     * 增加热度, 同时更新日榜, 周榜, 月榜
     *
     * @param videoId
     * @param score
     */
    public void incrScore(long videoId, double score) {
        incrScoreDaily(videoId, score);
        incrScoreWeek(videoId, score);
        incrScoreMonth(videoId, score);
    }


    /**
     * 增加每日排行榜 对应视频id的分数
     * 若更改后, 分数小于等于0, 则删除对应的视频id
     *
     * @param videoId
     * @param score
     */
    private Double incrScoreDaily(long videoId, double score) {
        return incrScoreSimple(DAILY_RANKING, videoId, score);
    }

    /**
     * 增加每周排行榜 对应视频id的分数
     * 若更改后, 分数小于等于0, 则删除对应的视频id
     *
     * @param videoId
     * @param score
     */
    private Double incrScoreWeek(long videoId, double score) {
        return incrScoreSimple(WEEK_RANKING, videoId, score);
    }

    /**
     * 增加每月排行榜 对应视频id的分数
     * 若更改后, 分数小于等于0, 则删除对应的视频id
     *
     * @param videoId
     * @param score
     */
    private Double incrScoreMonth(long videoId, double score) {
        return incrScoreSimple(MONTH_RANKING, videoId, score);
    }

    /**
     * 更改热度, 可能增加, 可能减少
     *
     * @param key
     * @param videoId
     * @param score
     * @return
     */
    private Double incrScoreSimple(String key, long videoId, double score) {
        ZSetOperations<String, Object> ops = redisTemplate.opsForZSet();
        // 查询视频对应的分区和时长
        Video videoInfo = videoService.getById(videoId);
        // 拼接分区id和视频id
        String value = "{" + videoInfo.getChannel() + "}" + "(" + videoId + ")";
        // 更改热度, 时长影响热度[[增量]], 减量不影响
        if (score > 0) {
            // log2 (0.5x + 1)
            score = score * Math.log(0.5 * videoInfo.getTotalTime() + 1)/ Math.log(2) + 1;
        }
        Double resultScore = ops.incrementScore(key, value, score);
        // 若减少热度后导致分数小于等于0, 则删除缓存
        if (resultScore != null && resultScore <= 0) {
            ops.remove(key, value);
        }
        return resultScore;
    }

    /**
     * 获取每日排行榜的前100的视频id以及热度分数
     */
    public Map<Long, Double> getRankDaily(int count) {
        return getRankMap(DAILY_RANKING, count);
    }

    public Map<Long, Double> getRankDaily(int count, int channelId) {
        return getRankMap(DAILY_RANKING, count, channelId);
    }

    /**
     * 获取每周排行榜的前100的视频id以及热度分数
     */
    public Map<Long, Double> getRankWeek(int count) {
        return getRankMap(WEEK_RANKING, count);
    }

    public Map<Long, Double> getRankWeek(int count, int channelId) {
        return getRankMap(WEEK_RANKING, count, channelId);
    }

    /**
     * 获取每月排行榜的前100的视频id以及热度分数
     */
    public Map<Long, Double> getRankMonth(int count) {
        return getRankMap(MONTH_RANKING, count);
    }

    public Map<Long, Double> getRankMonth(int count, int channelId) {
        return getRankMap(MONTH_RANKING, count, channelId);
    }


    private List<VideoRank> getRankList(String key, int count) {
        ZSetOperations<String, Object> ops = redisTemplate.opsForZSet();
        Set<ZSetOperations.TypedTuple<Object>> range = ops.reverseRangeWithScores(key, 0, count - 1);

        List<VideoRank> list = new ArrayList<>();
        if (range != null) {
            range.forEach(item -> {
                VideoRank rank = new VideoRank();
                rank.setVideoId(getVideoId(item));
                rank.setScoreFront(item.getScore());
                rank.setScoreEnd(item.getScore());
                list.add(rank);
            });
        }

        return list;
    }

    /**
     * 查询总排行榜
     *
     * @param key
     * @param count
     * @return
     */
    private Map<Long, Double> getRankMap(String key, int count) {
        ZSetOperations<String, Object> ops = redisTemplate.opsForZSet();
        Set<ZSetOperations.TypedTuple<Object>> range = ops.reverseRangeWithScores(key, 0, count - 1);

        Map<Long, Double> map = new LinkedHashMap<>();
        if (range != null) {
            range.forEach(item -> {
                // 截取value中的视频id
                Long videoId = getVideoId(item);
                map.put(videoId, item.getScore());
            });
        }

        return map;
    }

    /**
     * 查询指定分区下的排行榜
     *
     * @param key
     * @param count
     * @param channelId
     * @return
     */
    private Map<Long, Double> getRankMap(String key, int count, int channelId) {
        ZSetOperations<String, Object> ops = redisTemplate.opsForZSet();
        // 查询出指定分区id下的排行榜记录
        ScanOptions scanOptions = ScanOptions.scanOptions().match("*{" + channelId + "}*").build();
        Cursor<ZSetOperations.TypedTuple<Object>> result = ops.scan(key, scanOptions);

        Map<Long, Double> map = new LinkedHashMap<>();
        while (result.hasNext() && count-- > 0) {
            ZSetOperations.TypedTuple<Object> next = result.next();
            // 获取视频id
            Long videoId = getVideoId(next);

            map.put(videoId, next.getScore());
        }

        return map;
    }

    /**
     * 获取缓存的视频id
     *
     * @return
     */
    private Long getVideoId(ZSetOperations.TypedTuple<Object> item) {
        String value = (String) item.getValue();
        if (value == null) {
            throw new BusinessException("获取分区下排行榜时, 取得的value为null");
        }
        // 截取value中的视频id
        return Long.valueOf(value.substring(value.indexOf("(") + 1, value.indexOf(")")));
    }
}
